tiny grammar correction: Every message is sent

Grammar my old enemy - thanks for correction.

expects an acknowledgement for every message sent to Kinesis

Done.

Might be clearer as:
This throttling is controlled by the kinesis.{producer}.akka.max-outstanding-requests configuration value

Indeed much better.

just swap out send for sent :)

Done.

Couple of questions about this:

• If there's a break in sending messages (quiet period), will it complete and stop? As in, when outstandingMessages is empty.
• Do we need to store the outstanding messages, or could we instead defer to producer.outstandingRecordsCount() like the underlying actor does? I'm thinking about memory consumption here, albeit a small amount.
• When we receive a SendFailed - is failing the stage the right thing to do? As in, should we instead retry? Happy to defer this logic to another PR if it's worthwhile, in on the fence right now. Hard to know how often this will happen in real world usage. If often, then it's worth it.
• Regarding the earlier discussion about re-using the maxOutstandingRequests config. Am I correct in saying the following:
  • We'll stop requesting more messages when this count is hit (e.g. 1000 messages), which in theory should mean we won't actually throttle. So the fact that throttling is enabled won't matter because we won't start buffering messages within the actor.
  • The reason I implemented this was to reduce the number of futures in play (thereby saving memory and reducing work within the KPL). But it can potentially add latency depending on the value of throttling-retry-millis - so my question here is, do we ever want this to happen here, if not then it does actually make sense to use a different property - which goes against my earlier answer :)

• Q: If there's a break in sending messages (quiet period), will it complete and stop? As in, when outstandingMessages is empty.

  • A: No. The stream finishes only, if upstream finishes or the stream fails. See the SourceQueue Example added to the Readme : all messages send to the SourceQueue get published to Kinesis.

• Q: Do we need to store the outstanding messages, or could we instead defer to producer.outstandingRecordsCount() like the underlying actor does? I'm thinking about memory consumption here, albeit a small amount.

  • A: The graph stage runs async with the actor and the producer - the decision based on outstandingRecordsCount of the producer might be wrong. Since there is also no need to store a reference to all outstanding messages, I reworked my solution to only use an internal counter (and removed the map). This should be correct and has no memory impact.

• Q: When we receive a SendFailed - is failing the stage the right thing to do? As in, should we instead retry? Happy to defer this logic to another PR if it's worthwhile, in on the fence right now. Hard to know how often this will happen in real world usage. If often, then it's worth it.

  • A: Correct - the stage would fail in this case. I think adding a retry logic totally makes sense. I think this logic should be implemented in KinesisProducerActor as a separate PR. I created KinesisProducerActor: failed sends should be retried #50 to reflect this.

• Q: We'll stop requesting more messages when this count is hit (e.g. 1000 messages), which in theory should mean we won't actually throttle. So the fact that throttling is enabled won't matter because we won't start buffering messages within the actor.

  • A: akka streams are reactive: events are send downstream, demand is send upstream. The sink signals demand until maxOutstandingMessages are outstanding. If this number is reached no demand is signalled upstream. This also means that no more messages flow through the stream. The source of the stream will not emit more messages until demand is signalled.

• Q: The reason I implemented this was to reduce the number of futures in play (thereby saving memory and reducing work within the KPL). But it can potentially add latency depending on the value of throttling-retry-millis - so my question here is, do we ever want this to happen here, if not then it does actually make sense to use a different property - which goes against my earlier answer

  • A: If the source is faster than the sink, the number of outstanding messages will grow over time which will result in an application failure. IMHO it is necessary to define a maximum number of outstanding messages and to react (e.g. upstream should not produce more messages until demand is signalled) to prevent this. This is what akka streams provide out of the box. And this is also the reason to use a sensible throttling default. Since the number can be configured and can get arbitrary large, I don't think we will suffer from latency.

👍 thanks!

Good coverage of scenarios here!

One comment, some aspects might be simpler using the streams testkit - and it may avoid having to resolve the futureValue: https://doc.akka.io/docs/akka/2.5.5/scala/stream/stream-testkit.html#streams-testkit

For example using the TestSource to assert exceptions / completion and perhaps introducer breaks in the flow of messages more easily (to allow assertions between messages).

TestKit is great. The materialized Future produced by the sink is part of the GraphStage, that I wanted to test as well. If you see a test case not covered, that you would like to see - please let me know.